package site.jlopen.kuscia.rpc.impl;

import java.io.FileNotFoundException;
import java.util.Iterator;

import org.secretflow.v1alpha1.factory.KusciaAPIChannelFactory;
import org.secretflow.v1alpha1.kusciaapi.Job;
import org.secretflow.v1alpha1.kusciaapi.Job.ApproveJobRequest;
import org.secretflow.v1alpha1.kusciaapi.Job.ApproveJobResponse;
import org.secretflow.v1alpha1.kusciaapi.Job.BatchQueryJobStatusRequest;
import org.secretflow.v1alpha1.kusciaapi.Job.BatchQueryJobStatusResponse;
import org.secretflow.v1alpha1.kusciaapi.Job.CreateJobRequest;
import org.secretflow.v1alpha1.kusciaapi.Job.CreateJobResponse;
import org.secretflow.v1alpha1.kusciaapi.Job.DeleteJobRequest;
import org.secretflow.v1alpha1.kusciaapi.Job.DeleteJobResponse;
import org.secretflow.v1alpha1.kusciaapi.Job.QueryJobRequest;
import org.secretflow.v1alpha1.kusciaapi.Job.QueryJobResponse;
import org.secretflow.v1alpha1.kusciaapi.Job.RestartJobRequest;
import org.secretflow.v1alpha1.kusciaapi.Job.RestartJobResponse;
import org.secretflow.v1alpha1.kusciaapi.Job.StopJobRequest;
import org.secretflow.v1alpha1.kusciaapi.Job.StopJobResponse;
import org.secretflow.v1alpha1.kusciaapi.Job.WatchJobRequest;
import org.secretflow.v1alpha1.kusciaapi.JobServiceGrpc;
import org.secretflow.v1alpha1.kusciaapi.JobServiceGrpc.JobServiceBlockingStub;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import site.jlopen.kuscia.rpc.KusciaJobRpc;
import site.jlopen.kuscia.exception.SdkBaseException;
import site.jlopen.kuscia.properties.KusciaAPIProperties;

import io.grpc.ManagedChannel;

public class KusciaJobRpcImpl implements KusciaJobRpc {
	private static Logger logger = LoggerFactory.getLogger(KusciaJobRpcImpl.class);

	private JobServiceBlockingStub blockingStub;
	private ManagedChannel channel;

	public KusciaJobRpcImpl(KusciaAPIProperties properties) {
		try {
			KusciaAPIChannelFactory channelFactory = new KusciaAPIChannelFactory(properties.getAddress(),
					properties.getTokenFile(), properties.getTls(), properties.getProtocol());
			this.channel = channelFactory.newClientChannel();
			this.blockingStub = JobServiceGrpc.newBlockingStub(channel);
		} catch (Exception e) {
			if(e instanceof FileNotFoundException) {
				throw SdkBaseException.of("证书文件不存在", e);
			}
			throw SdkBaseException.of("Channel通讯异常", e);
		}
	}

	@Override
	public CreateJobResponse createJob(CreateJobRequest request) {
		try {
			CreateJobResponse response = blockingStub.createJob(request);
			checkResponse(response.getStatus());
			return response;
		} catch (Exception e) {
			logger.error("{}", e);
		} finally {
			if(null != channel) {
				channel.shutdownNow();
			}
		}
		return null;
	}

	@Override
	public QueryJobResponse queryJob(QueryJobRequest request) {
		try {
			QueryJobResponse response = blockingStub.queryJob(request);
			checkResponse(response.getStatus());
			return response;
		} catch (Exception e) {
			logger.error("{}", e);
		} finally {
			if(null != channel) {
				channel.shutdownNow();
			}
		}
		return null;
	}

	@Override
	public BatchQueryJobStatusResponse batchQueryJobStatus(BatchQueryJobStatusRequest request) {
		try {
			BatchQueryJobStatusResponse response = blockingStub.batchQueryJobStatus(request);
			//checkResponse(response.getStatus());
			return response;
		} catch (Exception e) {
			logger.error("{}", e);
		} finally {
			if(null != channel) {
				channel.shutdownNow();
			}
		}
		return null;
	}

	@Override
	public DeleteJobResponse deleteJob(DeleteJobRequest request) {
		try {
			DeleteJobResponse response = blockingStub.deleteJob(request);
			//checkResponse(response.getStatus());
			return response;
		} catch (Exception e) {
			logger.error("{}", e);
		} finally {
			if(null != channel) {
				channel.shutdownNow();
			}
		}
		return null;
	}

	@Override
	public StopJobResponse stopJob(StopJobRequest request) {
		try {
			StopJobResponse response = blockingStub.stopJob(request);
			//checkResponse(response.getStatus());
			return response;
		} catch (Exception e) {
			logger.error("{}", e);
		} finally {
			if(null != channel) {
				channel.shutdownNow();
			}
		}
		return null;
	}

	@Override
	public Iterator<Job.WatchJobEventResponse> watchJob(WatchJobRequest request) {
		try {
			Iterator<Job.WatchJobEventResponse> response = blockingStub.watchJob(request);
			return response;
		} catch (Exception e) {
			logger.error("{}", e);
		} finally {
			if(null != channel) {
				channel.shutdownNow();
			}
		}
		return null;
	}

	@Override
	public ApproveJobResponse approveJob(ApproveJobRequest request) {
		try {
			ApproveJobResponse response = blockingStub.approveJob(request);
			//checkResponse(response.getStatus());
			return response;
		} catch (Exception e) {
			logger.error("{}", e);
		} finally {
			if(null != channel) {
				channel.shutdownNow();
			}
		}
		return null;
	}

	@Override
	public RestartJobResponse RestartJob(RestartJobRequest request) {
		try {
			RestartJobResponse response = blockingStub.restartJob(request);
			//checkResponse(response.getStatus());
			return response;
		} catch (Exception e) {
			logger.error("{}", e);
		} finally {
			if(null != channel) {
				channel.shutdownNow();
			}
		}
		return null;
	}

}
